package cn.itcast.streaming.utils.test;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * Author itcast
 * Date 2021/9/28 8:50
 * Desc 用户下单以后，应该设置订单失效时间，用来提高用户的支付意愿如果用户下单15分钟未支付，则输出监控信息
 */
public class OrderTimeoutDemo {
    public static void main(String[] args) throws Exception {
        //1：初始化流式运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2：设置并行度为1
        env.setParallelism(1);

        //3：指定数据按照事件时间进行处理
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //4：构建数据源
        DataStream<OrderEvent> orderEventDataStream = env.fromCollection(
                Arrays.asList(
                        new OrderEvent(1, "create", 1558430842000L),//2019-05-21 17:27:22
                        new OrderEvent(2, "create", 1558430843000L), //2019-05-21 17:27:23
                        new OrderEvent(2, "other", 1558430845000L),  //2019-05-21 17:27:25
                        new OrderEvent(2, "pay", 1558430850000L), //2019-05-21 17:27:30
                        new OrderEvent(1, "pay", 1558431920000L) //2019-05-21 17:45:20
                )
        ).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<OrderEvent>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(OrderEvent element) {
                return element.getTimestamp();
            }
        });

        //5：设置模式规则
        Pattern<OrderEvent, OrderEvent> orderPayPattern = Pattern.<OrderEvent>begin("first")
                .where(new IterativeCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent orderEvent, Context context) throws Exception {
                        return orderEvent.getOrderStatus().equals("create");
                    }
                })
                .next("next")
                .where(new IterativeCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent orderEvent, Context context) throws Exception {
                        return orderEvent.getOrderStatus().equals("pay");
                    }
                })
                .within(Time.minutes(15));

        //6：将模式规则应用到数据流中
        PatternStream<OrderEvent> patternDataStream = CEP
                .pattern(orderEventDataStream.keyBy(OrderEvent::getOrderId), orderPayPattern);

        //7：定义侧输出流，获取超时的数据
        //默认情况下，符合规则的数据会直接输出打印出来，但是对于不符合条件的数据默认会丢弃掉，但是对于有些业务来说，不符合规则的数据也要进行二次处理
        OutputTag<OrderResult> orderTimeoutTag = new OutputTag<OrderResult>("orderTimeout") {
        };

        SingleOutputStreamOperator<OrderResult> resultDataStream = patternDataStream
                .select(orderTimeoutTag, new PatternTimeoutFunction<OrderEvent, OrderResult>() {
            @Override
            public OrderResult timeout(Map<String, List<OrderEvent>> pattern, long l) throws Exception {
                //用于处理订单超时的数据
                int orderId = pattern.getOrDefault("first", null).iterator().next().orderId;

                //返回超时支付的订单
                return new OrderResult(orderId, "超时支付");
            }
        }, new PatternSelectFunction<OrderEvent, OrderResult>() {
            @Override
            public OrderResult select(Map<String, List<OrderEvent>> pattern) throws Exception {
                //用于处理订单超时的数据
                int orderId = pattern.getOrDefault("first", null).iterator().next().orderId;

                //返回超时支付的订单
                return new OrderResult(orderId, "正常支付");
            }
        });

        //8：打印未超时的订单
        resultDataStream.print("正常完成支付的订单>>>");

        //9：打印已经超时支付的订单
        DataStream<OrderResult> timeoutDataStream = resultDataStream.getSideOutput(orderTimeoutTag);
        timeoutDataStream.printToErr("超时完成支付的订单>>>");

        // 10：运行任务
        env.execute();
    }

    //定义订单类
    public static class OrderEvent {
        public int orderId;
        public String orderStatus;
        public Long timestamp;

        public OrderEvent() {
        }

        public OrderEvent(int orderId, String orderStatus, Long timestamp) {
            this.orderId = orderId;
            this.orderStatus = orderStatus;
            this.timestamp = timestamp;
        }

        public int getOrderId() {
            return orderId;
        }

        public void setOrderId(int orderId) {
            this.orderId = orderId;
        }

        public String getOrderStatus() {
            return orderStatus;
        }

        public void setOrderStatus(String orderStatus) {
            this.orderStatus = orderStatus;
        }

        public Long getTimestamp() {
            return timestamp;
        }

        public void setTimestamp(Long timestamp) {
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "OrderEvent{" +
                    "orderId=" + orderId +
                    ", orderStatus='" + orderStatus + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }

    //定义订单结果类
    public static class OrderResult {
        public int orderId;
        public String payStatus;

        public OrderResult() {
        }

        public OrderResult(int orderId, String payStatus) {
            this.orderId = orderId;
            this.payStatus = payStatus;
        }

        public int getOrderId() {
            return orderId;
        }

        public void setOrderId(int orderId) {
            this.orderId = orderId;
        }

        public String getPayStatus() {
            return payStatus;
        }

        public void setPayStatus(String payStatus) {
            this.payStatus = payStatus;
        }

        @Override
        public String toString() {
            return "OrderResult{" +
                    "orderId=" + orderId +
                    ", payStatus='" + payStatus + '\'' +
                    '}';
        }
    }
}